package com.atguigu.gmall.common.rabbit.config;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.common.model.GmallCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ReturnCallback,RabbitTemplate.ConfirmCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 应用启动后触发一次:为IOC容器中RabbitTemplate模板对象设置生产者两个回调方法
     */
    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * 交换机确认:当消息发送到交换机将结果(成功或失败)异步发送给生产者
     * @param correlationData 回调相关性数据
     * @param ack 应答结果:true成功   false失败
     * @param cause 发送消息到交换机失败原因,失败才有值
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("[生产者确认之交换机确认],相关数据:{},回调结果:{},失败原因:{}",correlationData,ack,cause);
        if(ack){
            log.info("消息发送到交换机成功!");
        }else{
            log.error("消息发送到交换机失败,失败原因:{}",cause);
            //todo 进行消息重发保证稳定性
            GmallCorrelationData gmallCorrelationData = (GmallCorrelationData) correlationData;
            this.retrySendMsg(gmallCorrelationData);
        }
    }

    /**
     * 消息重试
     * @param gmallCorrelationData
     */
    private void retrySendMsg(GmallCorrelationData gmallCorrelationData) {
        //获取相关数据判断是否满足重发条件
        String uuid = gmallCorrelationData.getId();
        String gmallCorrelationDataStr = (String) redisTemplate.opsForValue().get(uuid);
        gmallCorrelationData = JSON.parseObject(gmallCorrelationDataStr, GmallCorrelationData.class);
        int retryCount = gmallCorrelationData.getRetryCount();
        if(retryCount >= 3){
            log.error("达到上限,将失败消息转入死信队列,会有单独消费者读取死信队列中消息:人工处理");
            return;
        }
        //调用模版进行消息重发一次
        System.out.println("消息重发");
        gmallCorrelationData.setRetryCount(gmallCorrelationData.getRetryCount() + 1);
        redisTemplate.opsForValue().set(uuid, JSON.toJSONString(gmallCorrelationData));
        redisTemplate.convertAndSend(gmallCorrelationData.getExchange(), gmallCorrelationData.getRoutingKey());
    }

    /**
     * 交换机确认之队列确认:交换机路由消息到队列失败才会将消息发送给生产者
     * @param message 返回消息对象
     * @param replyCode 应答码
     * @param replyText 错误原因
     * @param exchange 交换机名称
     * @param routingKey 路由键
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("队列确认失败,消息路由queue失败,消息={},应答码={},原因={},交换机={},路由键={}",message.toString(),replyCode,replyText,exchange,routingKey);
        //进行消息重发
        String redisKey = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        String str = (String) redisTemplate.opsForValue().get(redisKey);
        GmallCorrelationData gmallCorrelationData = JSON.parseObject(str, GmallCorrelationData.class);
        this.retrySendMsg(gmallCorrelationData);
    }
}